package com.sunday.common.mq.rocket.brave.study.apache.e3_Transactional_Message;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * https://github.com/apache/rocketmq-spring/wiki/FAQ
 * 如何发送事务性消息？
 *
 * 它在客户端需要两个步骤：
 *
 * a） 定义一个用@RocketMQTransactionListener注释的类，并实现 RocketMQLocalTransactionListener 接口，其中实现了 executeLocalTransaction（） 和 checkLocalTransaction（） 方法;
 *
 * b） 使用 RocketMQTemplate API 调用 sendMessageInTransaction（） 方法。
 *
 * 注意：在 rocketmq-spring 2.1.0 版本之后，注解@RocketmqTransactionListener无法设置 txProducerGroup、AK、SK。并且这些值与对应的火箭MQTemplate一致
 */
@Slf4j
@RocketMQTransactionListener
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private Random random = new Random();
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private String transactionId = "__transactionId__";
    private ConcurrentHashMap<Object, Integer> localTrans = new ConcurrentHashMap<>();

    /**
     * 0 -> UNKNOW              中间状态，它代表需要检查消息队列来确定状态。
     * 1 -> COMMIT_MESSAGE      提交事务，它允许消费者消费此消息。
     * 2 -> ROLLBACK_MESSAGE    回滚事务，它代表该消息将被删除，不允许被消费。
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // ... local transaction process, return bollback, commit or unknown
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        log.info("executeLocalTransaction TransactionId : {} , Status : {}", msg.getHeaders().get(transactionId), status);
        localTrans.put(msg.getHeaders().get(transactionId), status);
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // ... check transaction status and return bollback, commit or unknown
        // 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。
        Integer status = localTrans.get(msg.getHeaders().get(transactionId));
        if (null != status) {
            switch (status) {
                case 0:
                    log.info("checkLocalTransaction TransactionId : {} , old Status : {}", msg.getHeaders().get(transactionId), LocalTransactionState.UNKNOW);
                    if (random.nextBoolean()) {
                        return RocketMQLocalTransactionState.UNKNOWN;
                    } else {
                        log.info("checkLocalTransaction TransactionId : {} , new Status : {}", msg.getHeaders().get(transactionId), LocalTransactionState.COMMIT_MESSAGE);
                        return RocketMQLocalTransactionState.COMMIT;
                    }
                case 1:
                    log.info("checkLocalTransaction TransactionId : {} , Status : {}", msg.getHeaders().get(transactionId), LocalTransactionState.COMMIT_MESSAGE);
                    return RocketMQLocalTransactionState.COMMIT;
                case 2:
                    log.info("checkLocalTransaction TransactionId : {} , Status : {}", msg.getHeaders().get(transactionId), LocalTransactionState.ROLLBACK_MESSAGE);
                    return RocketMQLocalTransactionState.ROLLBACK;
            }
        }
        return RocketMQLocalTransactionState.COMMIT;
    }
}